package cloud.server1.service.impl;

import java.net.URL;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.netflix.hystrix.util.LongAdder;

import cloud.server1.service.QuasarService;
import co.paralleluniverse.fibers.Fiber;
import co.paralleluniverse.fibers.FiberExecutorScheduler;
import co.paralleluniverse.fibers.SuspendExecution;
import co.paralleluniverse.fibers.Suspendable;
import co.paralleluniverse.strands.Strand;
import co.paralleluniverse.strands.SuspendableRunnable;
import co.paralleluniverse.strands.concurrent.CountDownLatch;

@Service
public class QuasarServiceImpl implements QuasarService {
	/**
	 * 测试传统线程池和协程的效率
	 * 
	 * @return
	 */
	public String compare() {
		int count = 1000;
		try {
			testThreadpool(count);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		testFiber(count);
		return null;
	}

	private void testFiber(int count) {
		final CountDownLatch latch = new CountDownLatch(count);
		final LongAdder latency = new LongAdder();
		long t = System.currentTimeMillis();
		for (int i = 0; i < count; i++) {
			new Fiber<Void>("Caller", new SuspendableRunnable() {
				@Override
				public void run() throws SuspendExecution, InterruptedException {
					long start = System.currentTimeMillis();
					m1();
					start = System.currentTimeMillis() - start;
					latency.add(start);
					latch.countDown();
				}
			}).start();
		}
		try {
			latch.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		t = System.currentTimeMillis() - t;
		long l = latency.longValue() / count;
		System.out.println("fiber took: " + t + ", latency: " + l + " ms");
	}

	@Suspendable
	void m1() throws InterruptedException, SuspendExecution {
		String m = "m1";
		// System.out.println("m1 begin");
		m = m2();
		// System.out.println("m1 end");
		// System.out.println(m);
	}

	String m2() throws SuspendExecution, InterruptedException {
		String m = m3();
		Strand.sleep(1000);//模拟IO线程阻塞
		return m;
	}

	// or define in META-INF/suspendables
	@Suspendable
	String m3() {
		 List l = Stream.of(1, 2, 3).filter(i -> i % 2 ==
		 0).collect(Collectors.toList());
		return  l.toString();
	}

	void testThreadpool(int count) throws InterruptedException {
		final CountDownLatch latch = new CountDownLatch(count);
		ExecutorService es = Executors.newFixedThreadPool(200);
		LongAdder latency = new LongAdder();
		long t = System.currentTimeMillis();
		for (int i = 0; i < count; i++) {
			es.submit(() -> {
				long start = System.currentTimeMillis();
				try {
					m1();
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (SuspendExecution suspendExecution) {
					suspendExecution.printStackTrace();
				}
				start = System.currentTimeMillis() - start;
				latency.add(start);
				latch.countDown();
			});
		}
		latch.await();
		t = System.currentTimeMillis() - t;
		long l = latency.longValue() / count;
		System.out.println("thread pool took: " + t + ", latency: " + l + " ms");
		es.shutdownNow();
	}

	@Override
	public String recycle() {
		//设置1个线程，达到演示多个任务遇到IO阻塞时Quasar调度器将线程自动切换任务轮流运行
		ExecutorService es = Executors.newFixedThreadPool(1);
		FiberExecutorScheduler sc = new FiberExecutorScheduler("t", es);
		Fiber<Void> f1 = new Fiber<Void>("Caller1", sc, new SuspendableRunnable() {
			@Override
			public void run() throws SuspendExecution, InterruptedException {
				blockTask("Caller1");
			}
		}).start();
		Fiber<Void> f2 = new Fiber<Void>("Caller2", sc, new SuspendableRunnable() {
			@Override
			public void run() throws SuspendExecution, InterruptedException {
				blockTask("Caller2");
			}
		}).start();
		Fiber<Void> f3 = new Fiber<Void>("Caller3", sc, new SuspendableRunnable() {
			@Override
			public void run() throws SuspendExecution, InterruptedException {
				blockTask("Caller3");
			}
		}).start();
		return null;
	}

	public void blockTask(String taskName){
		for (int i = 0; i < 30; i++) {
			Thread cu = Thread.currentThread();
			try {
				//Strand.sleep(1000);//模拟IO线程阻塞
				new URL("http://www.baidu.com").openConnection().getContent();
				/*HttpGet request = new HttpGet("http://localhost:8771/kilim/sendAndReceive");
				// 获取当前客户端对象
				HttpClient httpClient = HttpClients.createDefault();
				// 通过请求对象获取响应对象
				HttpResponse response = httpClient.execute(request);
				// 判断网络连接状态码是否正常(0--200都数正常)
				if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
					String result = EntityUtils.toString(response.getEntity(), "utf-8");
					String rString = result;
				}*/
			} catch (Exception e) {
				e.printStackTrace();
			}
			System.out.println(taskName + ":运行位置  " + i + " 当前线程 " + cu.getId() + " 时间 " + new Date().toLocaleString());
		}
	}
}
